package com.batchengine.schedular;

import com.batchengine.cron.CronManagerUtil;
import com.batchengine.manager.ScheduledWorkerManager;
import com.batchengine.pool.TaskExecutorFactory;
import com.batchengine.pool.WorkerPoolFactory;

import static com.batchengine.cron.CronManagerUtil.getAllScheduledTasksForNextHour;
import static com.batchengine.cron.CronManagerUtil.getNextHour;
import static com.batchengine.pool.WorkerPoolFactory.createProducerConsumerPool;
import com.batchengine.task.RecurringJob;
import com.batchengine.task.ScheduledJob;
import static com.batchengine.task.TaskUtil.getScheduledJob;
import com.batchengine.util.EnvUtil;
import static com.batchengine.util.LoggingFactory.createLogger;
import com.batchengine.worker.*;
import org.apache.log4j.Logger;

import static java.lang.String.format;
import static java.lang.System.setProperty;
import java.util.*;
import static java.util.Calendar.getInstance;
import java.util.concurrent.ConcurrentHashMap;
import static java.util.concurrent.Executors.newScheduledThreadPool;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Created by  solutions
 * Author: sunil.incubator@yahoo.com
 * Date: Jul 22, 2009
 * <p/>
 * <p/>
 * //            if (null != job.getTaskInfo().getSignalClazz() && 0 < job.getTaskInfo().getSignalClazz().trim().length() && !"null".equalsIgnoreCase(job.getTaskInfo().getSignalClazz().trim())) {
 * //                PCStopSignal signalPC = (PCStopSignal) EnvUtil.createObjectz(job.getTaskInfo().getSignalClazz());
 * //                signalPC.setName(String.format("%s-killer", job.getInstanceName()));
 * //                workerPool.setSignal(signalPC);
 * //            } else {
 * //                workerPool.setSignal(new PCStopSignal(String.format("%s-killer", job.getInstanceName()), job.getTaskInfo().getTaskId(), workerPool, dataGatherer));
 * //            }
 */
public class CronSchedularWorkerManager extends ScheduledWorkerManager {

    private static final Logger logger = createLogger();

    private static final ScheduledExecutorService scheduler = newScheduledThreadPool(10);
    private static final ConcurrentHashMap<Long, ScheduledTask> cronTaskMap = new ConcurrentHashMap<Long, ScheduledTask>();
    private static final CronSchedularWorkerManager cronWorkerManager = new CronSchedularWorkerManager("ScheduledWorker-Manager");

    private CronSchedularWorkerManager(String name) {
        super(name);
    }

    @Override
    public Map<Long, ScheduledTask> getAllScheduledTasks() {
        return cronTaskMap;
    }

    public ScheduledTask getScheduledTask(Long jobId) {
        return cronTaskMap.get(jobId);
    }

    public static CronSchedularWorkerManager getCronSchedular() {
        return cronWorkerManager;
    }

    public void startSchedular() {
        Timer taskSchedular = new Timer();
        TimerTask schedularTask = new BatchSchedular(cronWorkerManager);
        taskSchedular.schedule(schedularTask, 1000, 1000);   //Todo change accordingly
    }

    public static class BatchSchedular extends ScheduledTask {

        private CronSchedularWorkerManager schedularWorkerManager;

        BatchSchedular(CronSchedularWorkerManager schedularWorkerManager) {
            super(901l);
            this.schedularWorkerManager = schedularWorkerManager;
        }

        @Override
        protected void doWork(Long scheduledJobId) {

            try {

                logger.debug("BatchEngine>rescheduling tasks");
                Collection<ScheduledJob> schedulredCrons = getAllScheduledTasksForNextHour(getNextHour());
                for (ScheduledJob job : schedulredCrons) {

                    if (null != cronTaskMap.get(job.getJobId())) {
                        continue;
                    }
                    ScheduledTask scheduledTask = new TaskSchedular(job.getJobId(), job.getInstanceName(), schedularWorkerManager);
                    scheduledTask.setScheduledTime(job.getScheduledTime());
                    cronTaskMap.put(job.getJobId(), scheduledTask);
                    scheduler.scheduleAtFixedRate(scheduledTask, job.getScheduledTime().getTime() - System.currentTimeMillis(), job.getScheduledTime().getTime(), TimeUnit.MILLISECONDS);
//                    scheduler.scheduleAtFixedRate(scheduledTask, getNextMin().getTime() - System.currentTimeMillis(), getNextMin().getTime(), TimeUnit.MILLISECONDS);
//                    Timer timer = new Timer();
//                    timer.schedule(scheduledTask, getNextMin());
//                    timer.schedule(scheduledTask, job.getScheduledTime());
                    logger.debug(format("BatchEngine>sceduled task:%d @ %s", job.getJobId(), job.getScheduledTime()));
                }
            } catch (Exception e) {
                logger.error("Error#CronSchedularWorkerManager@doWork", e);
            }
        }
    }

    public static class TaskSchedular extends ScheduledTask {

        private CronSchedularWorkerManager schedularWorkerManager;

        TaskSchedular(Long jobId, String name, CronSchedularWorkerManager schedularWorkerManager) {
            super(jobId, name);
            this.schedularWorkerManager = schedularWorkerManager;
            logger.debug("scheduled this task for " + scheduledExecutionTime() + this);
        }

        @Override
        protected void doWork(Long scheduledJobId) throws InterruptedException {

            if (100 < scheduledJobId) {
                return;
            }

            ScheduledJob job = getScheduledJob(scheduledJobId);
            if (null == job.getTaskInfo().getSelctorClazz() || 0 == job.getTaskInfo().getSelctorClazz().trim().length()) {
                createExecutorPool(job);
                return;
            }
            createPCPool(job);
        }

        private void createExecutorPool(RecurringJob job) {
            workerPool = WorkerPoolFactory.createExecutorPool(job.getMaxWorkers(), job.getMaxWorkers(), job.getTaskInfo().getTaskId(), new TaskExecutorFactory());
            workerPool.setManage(false);
            workerPool.setDelaySecs(job.getDelay());
            workerPool.setName(format("%s", job.getInstanceName()));
            schedularWorkerManager.addWorkerPool(workerPool);
            workerPool.startWorkers();
        }

        private void createPCPool(RecurringJob job) {

            WorkSelector workSelector = (WorkSelector) EnvUtil.createObjectz(job.getTaskInfo().getSelctorClazz());
            DataGatherer dataGatherer = new ScheduledDataGatherer(format("%s-Gatherer", job.getInstanceName()), job.getJobId(), workSelector);
            workerPool = createProducerConsumerPool(job.getMaxWorkers(), job.getMaxWorkers(), job.getTaskInfo().getTaskId(), new ConsumerFactory(workSelector.getWorkQ(), true), dataGatherer);
            workerPool.setName(job.getInstanceName());
            workerPool.setManage(false);
            workerPool.setDelaySecs(job.getDelay());
            schedularWorkerManager.addWorkerPool(workerPool);
            workerPool.startWorkers();
        }
    }

    protected void cleanUp() {
        scheduler.shutdownNow();
    }

    public static void main(String[] args) throws InterruptedException {

//        setProperty("conf.file", "/host//codebase/bashengine/settings/conf/be.properties");
        Timer taskSchedular = new Timer();
        ScheduledTask scheduledTask1 = new TaskSchedular(1l, "123", new CronSchedularWorkerManager("testing-schedularWorkerManager"));
        ScheduledTask scheduledTask2 = new TaskSchedular(2l, "321", new CronSchedularWorkerManager("testing-schedularWorkerManager"));
        Date min = getNextMin();
//        taskSchedular.schedule(scheduledTask1, min);
//        taskSchedular.schedule(scheduledTask2, min);

        long curr = System.currentTimeMillis();
        long diff = min.getTime() - curr;

        System.out.println(min.getTime() + " - " + curr + " = " + diff);

        scheduler.scheduleAtFixedRate(scheduledTask1, diff, min.getTime(), TimeUnit.MILLISECONDS);
    }

    private static Date getNextMin() {
        Calendar c = getInstance();
        c.set(Calendar.SECOND, 0);
        c.set(Calendar.MINUTE, c.get(Calendar.MINUTE) + 1);
        return c.getTime();
    }
}